#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Date    : Mon Jul 22 2019
# @Author  : xujunfu (xjf891121@gmail.com})
# @Link    : http://junfuxu.github.io/
# -*-coding:utf-8 -*-
import json

from kafka import KafkaConsumer

from . import config, utils
from .handler.ed import EDHandler
from .handler.gps import GPSHandler
from .handler.online import OnlineHandler


class KYReceiver():
    def __init__(self, *topics):
        self.logger = utils.get_logger('consumer-logger')
        topics = topics if topics else config.KAFKA['TOPICS']
        self.consumer = KafkaConsumer(
            *topics,
            group_id=config.KAFKA['GROUP_ID'],
            bootstrap_servers=config.KAFKA['SERVERS'],
            auto_offset_reset='latest',
            value_deserializer=lambda m: json.loads(m.decode('ascii'))
        )
        self.consumer.subscribe(topics)
        self.running = False

    def run(self):
        self.running = True
        self.ed_handler = EDHandler()
        self.online_handle = OnlineHandler()
        self.gps_handler = GPSHandler()
        self.logger.info('start consumer')
        for msg in self.consumer:
            if msg.value['deviceId'] in ['866275036350446']:
                continue
            try:
                if msg.topic == 'truck_gps_data':
                    self.gps_handler.handle(msg.value)
                elif msg.topic == "truck_ed_data":
                    self.ed_handler.handle(msg.value)
                elif msg.topic == "truck_online_data":
                    self.online_handle.handle(msg.value)
                else:
                    print("错误信息", msg)
            except Exception as e:
                self.logger.exception(e)
